package com.winit.websocket.handler;

import com.alibaba.fastjson.JSONObject;
import com.winit.mq.broker.MessageBroker;
import com.winit.mq.listener.QueueListener;
import com.winit.mq.message.Message;
import com.winit.mq.task.TailLogTask;
import com.winit.util.Constants;
import com.winit.util.SysInfoTable;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import org.springframework.web.socket.*;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * websocket消息处理类
 * @author wangjian
 *
 */
@Component
public class MyHandler implements WebSocketHandler {


	//	protected MessageBroker messageBroker= SpringContextHolder.getBean(MessageBroker.class);;
	private Map<WebSocketSession,List<QueueListener>> sessionToListenerMap=new ConcurrentHashMap();
	private static final Logger logger = LoggerFactory.getLogger(MyHandler.class);
	private static Map<String,String> buildTaskMap = new HashMap();

	private MessageBroker messageBroker=MessageBroker.getInstance();


	@Override
	public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus)
			throws Exception {
		logger.info("关闭ws消息监听......");
		List<QueueListener> listeners=sessionToListenerMap.get(session);
		if(sessionToListenerMap.get(session)!=null){
			sessionToListenerMap.remove(session);
		}
		if(listeners!=null && listeners.size()!=0) {
			messageBroker.removeListener(listeners);
		}
		this.closeSocket(session);
	}

	@Override
	public void afterConnectionEstablished(WebSocketSession session)
			throws Exception {
		logger.info("connect to the websocket success......");
		SysInfoTable.websocketList.add(session);
	}


	@Override
	public void handleMessage(WebSocketSession session, WebSocketMessage<?> message)
			throws Exception {
		String messag = (String) message.getPayload();

		if(StringUtils.isEmpty(messag)){
			return;
		}
		List<String> queueNames = new ArrayList<String>();

//		if(StringUtils.isNoneEmpty(json.getString("userId"))){
//			queueNames.add(json.getString("userId"));
//		}
		queueNames.add(messag);
		for(String qName:queueNames){
			final String queueName=qName;
			// 基于消息队列的Websocket监听器,用于监听消息队列, 并试图向ws的session推送消息.
			logger.info("创建ws消息监听,queueName="+queueName);
			registerListener(session, queueName);
		}
	}
	protected void registerListener(final WebSocketSession webSocketSession, final String queueName) {
		//创建监听
		final QueueListener listener=new QueueListener() {

			@Override
			public void onMessage(Message message) {
				try {
					//检查连接是否开启 开启时发送消息
					if (webSocketSession.isOpen()){
						logger.info("队列["+queueName+"]消息发送开始.");
						WebSocketMessage wsMessage=new TextMessage(JSONObject.toJSONString(message));
						webSocketSession.sendMessage(wsMessage);
						logger.info("队列["+queueName+"]消息发送结束.");
					}

				}catch (Exception e){
					logger.error("队列["+queueName+"]消息发送失败."+JSONObject.toJSONString(message),e) ;
				}
			}
		};
		//绑定session和listener的关系
		List<QueueListener> listeners=sessionToListenerMap.get(webSocketSession);
		if(listeners==null){
			listeners=new ArrayList();
			sessionToListenerMap.put(webSocketSession,listeners);
		}
		listeners.add(listener);
		//绑定queue和listener的关系
		String name=queueName;
		messageBroker.addListener(name,listener,true);

	}

	@Override
	public void handleTransportError(WebSocketSession session, Throwable exception)
			throws Exception {
		logger.info("websocket connection closed......");
		if(session.isOpen()){
			session.close();
		}
		this.closeSocket(session);
	}

	@Override
	public boolean supportsPartialMessages() {
		return false;
	}


	private void closeSocket(WebSocketSession session){
		SysInfoTable.websocketList.remove(session);
		String threadId = (String) session.getAttributes().get(Constants.WEBSOCKET_SESSION_THREADID_INDEX);
		if(!StringUtils.isEmpty(threadId)){
			TailLogTask task = SysInfoTable.threadMap.get(threadId);
			logger.info("关闭tail线程，threadId="+threadId+";task="+task);
			if(task != null){
				task.shutThread();
			}
			session.getAttributes().remove(Constants.WEBSOCKET_SESSION_THREADID_INDEX);
			SysInfoTable.threadMap.remove(threadId);
		}
	}
}
